package com.blueline.sms.webservice.impl;

import com.blueline.sms.webservice.ISender;
import com.blueline.util.concurrent.BlockingFileQueue;


import javax.jws.WebService;

import java.io.*;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

@WebService(endpointInterface = "com.blueline.sms.webservice.ISender")
public class SenderImpl implements ISender {

    final BlockingQueue<byte[]> queue ;
    Thread takeThread;
    Timer showtime;

    long lastCount=0;
    long lastPCount=0;

    long lastIOw=0;
    long lastIOr=0;

    AtomicLong iow=new AtomicLong(0);
    AtomicLong ior=new AtomicLong(0);

    AtomicLong count=new AtomicLong(0);
    AtomicLong pcount=new AtomicLong(0);

    AtomicLong cleanCount=new AtomicLong(0);

    public SenderImpl() throws Exception {
        queue = new BlockingFileQueue("db", 1024 * 1024 * 64);

        showtime=new Timer();
        showtime.schedule(new TimerTask() {
            @Override
            public void run() {
                long tempCount=count.get();
                long tempPcount=pcount.get();

                long tempiow=iow.get();
                long tempior=ior.get();

                long now=System.currentTimeMillis();

                if(lastCount==tempCount && lastPCount==tempPcount && lastIOw==tempiow && lastIOr==tempior){
                    if (cleanCount.incrementAndGet()==10) {
                        count.set(0);
                        pcount.set(0);
                        lastCount=0;
                        lastPCount=0;
                        tempCount=0;
                        tempPcount=0;
                        iow.set(0);
                        ior.set(0);
                        lastIOw=0;
                        lastIOr=0;
                        tempiow=0;
                        tempior=0;
                    }
                }else{
                    cleanCount.set(0);
                }
                System.out.println(now+":" + tempCount+"/"+tempPcount+"\t "+(tempPcount-lastPCount)+"/Sec p\t"+(tempCount-lastCount)+"/sec c\t\t"+ tempiow+"/"+tempior+"\t "+(tempiow-lastIOw)+"/Sec w\t"+(tempior-lastIOr)+"/sec r");

                lastCount=tempCount;
                lastPCount=tempPcount;

                lastIOw=tempiow;
                lastIOr=tempior;


            }
        }, 1000, 1000);

        takeThread=new Thread(new Runnable() {
            @Override
            public void run() {
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        byte[] data=queue.take();
                        ior.addAndGet(data.length);
                        count.incrementAndGet();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        takeThread.setDaemon(true);
        takeThread.start();
    }

    @Override
    public String mongateCsSpSendSmsNew(String userId, String password, String pszMobis, String pszMsg, int iMobiCount, String pszSubPort) {
        try{
            String [] mobis=pszMobis.split(",");
            long time= System.currentTimeMillis();

            byte[] data=enCode(time,userId,mobis,pszMsg,iMobiCount,pszSubPort);
            iow.addAndGet(data.length);
            queue.put(data);
            pcount.incrementAndGet();

            return String.valueOf(time);
        }catch (Exception e){
            return "-999";
        }
    }

    public static byte[] enCode(long id,String userId,String pszMobis, String pszMsg, String pszSubPort) throws UnsupportedEncodingException {
        String time=String.valueOf(id);
        return (userId + ":" + pszMobis + ":" + pszMsg + ":" + pszSubPort + ":" + time).getBytes("UTF-8");
    }

    public static byte[] enCode(long id,String userId,String [] pszMobis, String pszMsg, int iMobiCount, String pszSubPort) throws IOException {
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(3*1024);
        try {

            DataOutputStream outputStreamWriter = new DataOutputStream(byteArrayOutputStream);
            outputStreamWriter.writeLong(id);
            outputStreamWriter.writeInt(Integer.valueOf(userId));
            outputStreamWriter.writeInt(Integer.valueOf(pszSubPort));
            outputStreamWriter.writeByte(iMobiCount);
            for (int i = 0; i < iMobiCount; i++) {
                outputStreamWriter.writeLong(Long.valueOf(pszMobis[i]));
            }
            byte[] msgbytes = pszMsg.getBytes("UTF-8");
            outputStreamWriter.writeInt(msgbytes.length);
            outputStreamWriter.write(msgbytes);
            outputStreamWriter.flush();

            return byteArrayOutputStream.toByteArray();
        }finally {
            byteArrayOutputStream.close();
        }
    }
}
